package cn.lsh.hbase.coprocessor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.ByteArrayComparable;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;

/**
 * 观察者协处理器，类似于mysql的触发器，采用切面编程，在定义的一系列操作前后执行用户设置处理逻辑
 */
public class RegionObserverCoprocessor implements RegionCoprocessor, RegionObserver {
	Logger log = LoggerFactory.getLogger(RegionObserverCoprocessor.class);

	/**协处理器是运行于region中的，每一个region都会加载协处理器*/
	private RegionCoprocessorEnvironment env = null;

	private static final byte[] TABLE = Bytes.toBytes("users");

	private static final byte[] CF = Bytes.toBytes("salaryDet");

	private static final byte[] CF2 = Bytes.toBytes("personalDet");

	private static final byte[] ADMIN = Bytes.toBytes("admin");

	@Override
	public void start(CoprocessorEnvironment env) throws IOException {
		//如果您从HBase -site.xml加载一个协处理器，然后使用HBase Shell再次加载相同的协处理器，那么它将被第二次加载。
		// 同一个类将存在两次，第二个实例将具有更高的ID(因此优先级较低)。其效果是，重复的协处理器被有效地忽略。
		this.env = (RegionCoprocessorEnvironment) env;
	}

	@Override
	public void stop(CoprocessorEnvironment env) throws IOException {

	}

	@Override
	public Optional<RegionObserver> getRegionObserver() {
		return Optional.of(this);
	}

	@Override
	public void preGetOp(ObserverContext<RegionCoprocessorEnvironment> c, final Get get, final List<Cell> result) throws IOException {
		log.info("---Get操作前置处理 start---");
		//get操作前置处理
		if (Bytes.equals(get.getRow(), ADMIN)) {
			result.add(new KeyValue(get.getRow(), Bytes.toBytes("错误"), Bytes.toBytes("详细"), Bytes.toBytes("您不能获取管理员信息")));
			//禁止访问，直接返回错误信息
			c.bypass();
		} else if (Bytes.equals(get.getRow(), Bytes.toBytes("jverne"))) {
			//增加查询条件
			get.addFamily(CF);
			//增加一条记录
			KeyValue kv = new KeyValue(get.getRow(), CF2, Bytes.toBytes("addPre"), Bytes.toBytes("boss"));
			result.add(kv);
		}
		log.info("---Get操作前置处理 end---");

	}

	@Override
	public void postGetOp(ObserverContext<RegionCoprocessorEnvironment> c, Get get, List<Cell> result) throws IOException {
		log.info("---Get操作后置处理 start---");
		if (Bytes.equals(get.getRow(), Bytes.toBytes("cdickens"))) {
			//修改查询条件，应该无效
			get = new Get(Bytes.toBytes("jverne"));
			//增加一列记录
			KeyValue kv = new KeyValue(get.getRow(), CF2, Bytes.toBytes("addPre"), Bytes.toBytes("coprocessor"));
			result.add(kv);
			c.bypass();
		}
		log.info("---Get操作后置处理 end---");
	}

	@Override
	public void prePut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {
		log.info("---Put操作前置处理 start---");
		if (Bytes.equals(ADMIN, put.getRow())) {
			//对于admin用户，存在就不覆盖
			c.bypass();
		}
		log.info("---Put操作前置处理 end---");
	}

	@Override
	public void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit, Durability durability) throws IOException {
		log.info("---Put操作后置处理 start---");

		log.info("---Put操作后置处理 end---");
	}

	@Override
	public boolean preCheckAndPut(ObserverContext<RegionCoprocessorEnvironment> c, byte[] row, byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator, Put put, boolean result) throws IOException {
		return false;
	}

	@Override
	public boolean postCheckAndPut(ObserverContext<RegionCoprocessorEnvironment> c, byte[] row, byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator, Put put, boolean result) throws IOException {
		return false;
	}

	@Override
	public void preDelete(ObserverContext<RegionCoprocessorEnvironment> c, Delete delete, WALEdit edit, Durability durability) throws IOException {
		log.info("---Delete操作前置处理 start---");
		if (Bytes.equals(ADMIN, delete.getRow())) {
			throw new IOException("不能删除ADMIN用户");
		}
		log.info("---Delete操作前置处理 end---");
	}

	@Override
	public void postDelete(ObserverContext<RegionCoprocessorEnvironment> c, Delete delete, WALEdit edit, Durability durability) throws IOException {
		log.info("---Delete操作后置处理 start---");
		log.info("---Delete操作后置处理 end---");
	}

	@Override
	public boolean preCheckAndDelete(ObserverContext<RegionCoprocessorEnvironment> c, byte[] row, byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator, Delete delete, boolean result) throws IOException {
		return false;
	}

	@Override
	public boolean postCheckAndDelete(ObserverContext<RegionCoprocessorEnvironment> c, byte[] row, byte[] family, byte[] qualifier, CompareOperator op, ByteArrayComparable comparator, Delete delete, boolean result) throws IOException {
		return false;
	}

	@Override
	public boolean preExists(ObserverContext<RegionCoprocessorEnvironment> c, Get get, boolean exists) throws IOException {
		return false;
	}

	@Override
	public boolean postExists(ObserverContext<RegionCoprocessorEnvironment> c, Get get, boolean exists) throws IOException {
		return false;
	}

	@Override
	public void preScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Scan scan) throws IOException {
		log.info("---Scan开启操作前置处理 start---");
		//添加一个过滤器，不允许扫描rowKey为ADMIN的行的信息
		Filter filter = new RowFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(ADMIN));
		scan.setFilter(filter);
		log.info("---Scan开启操作前置处理 end---");
	}

	@Override
	public RegionScanner postScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Scan scan, RegionScanner s) throws IOException {
		log.info("---Scan开启操作后置处理 start---");
		log.info("---Scan开启操作后置处理 end---");
		//注意在这里不能反回null
		return s;
	}

	@Override
	public void preScannerClose(ObserverContext<RegionCoprocessorEnvironment> c, InternalScanner s) throws IOException {

	}

	@Override
	public boolean preScannerNext(ObserverContext<RegionCoprocessorEnvironment> c, InternalScanner s, List<Result> result, int limit, boolean hasNext) throws IOException {
		return hasNext;
	}

	@Override
	public boolean postScannerNext(ObserverContext<RegionCoprocessorEnvironment> c, InternalScanner s, List<Result> result, int limit, boolean hasNext) throws IOException {
		log.info("---ScanNext操作后置处理 start---");
		//从扫描结果汇总移除ADMIN信息
		result.removeIf(r -> Bytes.equals(ADMIN, r.getRow()));
		//虽然preScannerOpen添加了过滤器，但是有可能被其他过滤器覆盖掉，所以可以在这里显示的移除ADMIN信息
		log.info("---ScanNext操作后置处理 end---");
		return hasNext;
	}
}
